package com.ferry.lock.engine.zookeeper.menagerie.locks;

import com.ferry.lock.engine.zookeeper.menagerie.ZKSessionManager;
import com.ferry.lock.engine.zookeeper.menagerie.ZKUtils;
import com.ferry.lock.lock.ILock;
import com.ferry.lock.lock.IReentrantReadWriteLock;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * Created by daiyong
 */
public class ReentrantZKReadWriteLock implements IReentrantReadWriteLock {

	private static final String readLockPrefix = "readLock";

	private static final String writeLockPrefix = "writeLock";

	private final ReadLock readLock;

	private final WriteLock writeLock;

	public ReentrantZKReadWriteLock(String baseNode, ZKSessionManager zkSessionManager) {
		this(baseNode,zkSessionManager, ZooDefs.Ids.OPEN_ACL_UNSAFE);
	}

	public ReentrantZKReadWriteLock(String baseNode, ZKSessionManager zkSessionManager, List<ACL> privileges) {
		readLock = new ReadLock(baseNode, zkSessionManager, privileges);
		writeLock = new WriteLock(baseNode, zkSessionManager, privileges);
	}

	@Override
	public ILock readLock() {
		return readLock;
	}

	@Override
	public ILock writeLock() {
		return writeLock;
	}

	public class ReadLock extends ReentrantZKLock{

		private ReadLock(String baseNode, ZKSessionManager zkSessionManager, List<ACL> privileges) {
			super(baseNode, zkSessionManager, privileges);
		}

		@Override
		protected boolean tryAcquireDistributed(ZooKeeper zk, String lockNode, boolean watch) throws KeeperException, InterruptedException {
			//获取所有的写锁
			List<String> writeLocks = ZKUtils.filterByPrefix(zk.getChildren(baseNode, false), writeLockPrefix);

			ZKUtils.sortBySequence(writeLocks, lockDelimiter);

			long mySequenceNumber = ZKUtils.parseSequenceNumber(lockNode,lockDelimiter);
			//get all write Locks which are less than my sequence number
			List<String> aheadWriteLocks = new ArrayList<String>();
			for(String writeLock:writeLocks){
				long lockSeqNbr = ZKUtils.parseSequenceNumber(writeLock, lockDelimiter);
				if(lockSeqNbr <mySequenceNumber){
					//a write lock ahead of us, so add it in
					aheadWriteLocks.add(writeLock);
				}
			}
            /*
            since aheadWriteLocks is in sorted order, we know that the last element which is still in ZooKeeper
            is the write lock that we have to wait for
             */
			while(aheadWriteLocks.size()>0){
				String lastWriteLock = aheadWriteLocks.remove(aheadWriteLocks.size() - 1);
				Stat stat;
				if(watch){
					stat = zk.exists(baseNode+"/"+ lastWriteLock,signalWatcher);
				}else
					stat = zk.exists(baseNode+"/"+ lastWriteLock, false);
				if(stat!=null){
					//this node still exists, so wait in line behind it
					return false;
				}
			}
			return true;
		}

		@Override
		protected String getBaseLockPath() {
			return baseNode + "/" + readLockPrefix + lockDelimiter;
		}

		@Override
		public Condition newCondition() {
			throw new UnsupportedOperationException("Read Locks do not support conditions");
		}
	}

	public class WriteLock extends ReentrantZKLock{

		private WriteLock(String baseNode, ZKSessionManager zkSessionManager, List<ACL> privileges) {
			super(baseNode, zkSessionManager, privileges);
		}

		@Override
		protected boolean tryAcquireDistributed(ZooKeeper zk, String lockNode, boolean watch) throws KeeperException, InterruptedException {
			List<String> locks = ZKUtils.filterByPrefix(zk.getChildren(baseNode, false), readLockPrefix, writeLockPrefix);
			ZKUtils.sortBySequence(locks,lockDelimiter);

			long mySequenceNumber = ZKUtils.parseSequenceNumber(lockNode,lockDelimiter);
			List<String> aheadLocks = new ArrayList<String>();
			for(String lock: locks){
				if(ZKUtils.parseSequenceNumber(lock,lockDelimiter)<mySequenceNumber){
					aheadLocks.add(lock);
				}
			}

			while(aheadLocks.size()>0){
				String lastReadLock = aheadLocks.remove(aheadLocks.size() - 1);
				Stat stat;
				if(watch)
					stat = zk.exists(baseNode + "/"+ lastReadLock, signalWatcher);
				else
					stat = zk.exists(baseNode + "/"+ lastReadLock, false);
				if (stat != null){
					return false;
				}
			}
			return true;
		}


		@Override
		protected String getBaseLockPath() {
			return baseNode+"/"+writeLockPrefix+lockDelimiter;
		}
	}
}
